package com.yy.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableRabbit
public class RabbitConfig {

    @Bean(name = "rabbitMessageListenerContainer")
    public DirectMessageListenerContainer directMessageListenerContainer(ConnectionFactory connectionFactory) {
        DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setPrefetchCount(5);
        container.setConsumersPerQueue(5);
        container.setMessagesPerAck(1);
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        container.setTaskExecutor(executor);
        return container;
    }


    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }


    @Bean("orderQueue")
    public Queue orderQueue() {
        return QueueBuilder.durable("q.order").build();
    }

    @Bean
    public Queue ttlQueue() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-message-ttl", 10000);
        args.put("x-dead-letter-exchange", "x.dlx");
        args.put("x-dead-letter-routing-key", "k.dlx");
        Queue queue = new Queue("q.ttl", true, false, false, args);
        return queue;
    }

    @Bean("orderExchange")
    public Exchange orderExchange() {
        return new DirectExchange("x.order", true, false, null);
    }


    //延迟队列交换器
    @Bean
    public Exchange ttlExchange() {
        return new DirectExchange("x.ttl", true, false, null);
    }


    // 死信队列
    @Bean
    public Queue dlxQueue() {
        Map<String, Object> args = new HashMap<>();
        Queue dlq = new Queue("q.dlx", true, false, false, args);

        return dlq;
    }

    // 死信队列交换器
    @Bean
    public Exchange dlxExchange() {
        return new DirectExchange("x.dlx", true, false, null);
    }

    // 绑定 订单队列和交换器
    @Bean
    public Binding orderBinding(@Qualifier("orderQueue") Queue orderQueue, @Qualifier("orderExchange") Exchange orderExchange) {
        return BindingBuilder.bind(orderQueue).to(orderExchange).with("k.order").noargs();
    }

    // 绑定 延迟队列和交换器
    @Bean
    public Binding ttlBinding(@Qualifier("ttlQueue") Queue ttlQueue, @Qualifier("ttlExchange") Exchange ttlExchange) {
        return BindingBuilder.bind(ttlQueue).to(ttlExchange).with("k.ttl").noargs();
    }

    // 绑定 死信队列和交换器
    @Bean
    public Binding dlxBinding(@Qualifier("dlxQueue") Queue dlxQueue, @Qualifier("dlxExchange") Exchange dlxExchange) {
        return BindingBuilder.bind(dlxQueue).to(dlxExchange).with("k.dlx").noargs();
    }

}
